package org.qy.star_server.data.center;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.qy.star_common.enums.ExchangeType;
import org.qy.star_common.message.Message;
import org.qy.star_server.core.*;
import org.qy.star_server.core.Queue;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Repository;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author yinger
 * @description
 * @date 2024/3/28 17:15
 **/
@Data
@Slf4j
public class MemoryDataCenter {
    public static MemoryDataCenter memory = null;

    private MemoryDataCenter() {

    }

    public static MemoryDataCenter getMemoryBean() {
        if (memory == null) { // 第一次检查
            synchronized (MemoryDataCenter.class) {
                if (memory == null) { // 第二次检查
                    memory = new MemoryDataCenter();
                }
            }
        }
        return memory;
    }

    /************************************************** 内存元数据 **************************************************/
    // 消息中心，存储所有消息 <id, message>
    private final ConcurrentHashMap<String, Message> messageCenter = new ConcurrentHashMap<>();
    // 用户中心，存储所有用户
    private final ConcurrentHashMap<User.UserGloballyUniqueIdentity, User> userCenter = new ConcurrentHashMap<>();
    // 虚拟机中心，存储所有虚拟机
    private final ConcurrentHashMap<VirtualHost.VirtualHostGloballyUniqueIdentity, VirtualHost> virtualHostCenter = new ConcurrentHashMap<>();
    // 交换机中心，存储所有交换机
    private final ConcurrentHashMap<Exchange.ExchangeGloballyUniqueIdentity, Exchange> exchangeCenter = new ConcurrentHashMap<>();
    // 队列中心，存储所有队列
    private final ConcurrentHashMap<Queue.QueueGloballyUniqueIdentity, Queue> queueCenter = new ConcurrentHashMap<>();
    // 绑定中心，存储所有绑定关系
    private final ConcurrentHashMap<Binding.BindingeGloballyUniqueIdentity, Binding> bindingCenter = new ConcurrentHashMap<>();
    // 队列消息中心，存储所有队列消息
    private final ConcurrentHashMap<Queue.QueueGloballyUniqueIdentity, LinkedList<Message>> queueMessageCenter = new ConcurrentHashMap<>();
    // 待确认消息中心，存储所有待确认消息
    private final ConcurrentHashMap<Queue.QueueGloballyUniqueIdentity, ConcurrentHashMap<String, Message>> waitAckMessageCenter = new ConcurrentHashMap<>();

    /************************************************** ##### *****************************************************/
    public void insertUser(String username, String password) {
        try {
            User user = new User(username, password);
            User.UserGloballyUniqueIdentity userGloballyUniqueIdentity = new User.UserGloballyUniqueIdentity(username);
            userCenter.put(userGloballyUniqueIdentity, user);
        } catch (Exception e) {
            log.error("insertUser error: " + e.getMessage());
        }
    }

    public void deleteUser(String username) {
        try {
            User.UserGloballyUniqueIdentity userGloballyUniqueIdentity = new User.UserGloballyUniqueIdentity(username);
            User user = userCenter.get(userGloballyUniqueIdentity);
            if (user == null) {
                // 待删除用户不存在
                return;
            }
            userCenter.remove(userGloballyUniqueIdentity);
        } catch (Exception e) {
            log.error("deleteUser error: " + e.getMessage());
        }
    }

    public User selectUser(String username) {
        try {
            User.UserGloballyUniqueIdentity userGloballyUniqueIdentity = new User.UserGloballyUniqueIdentity(username);
            User user = userCenter.get(userGloballyUniqueIdentity);
            if (user == null) {
                // 待查询用户不存在
                return null;
            }
            return user;
        } catch (Exception e) {
            return null;
        }
    }

    public void updateUser(String username, String newPassword) {
        try {
            User user = new User(username, newPassword);
            User.UserGloballyUniqueIdentity userGloballyUniqueIdentity = new User.UserGloballyUniqueIdentity(username);
            userCenter.put(userGloballyUniqueIdentity, user);
        } catch (Exception e) {
            log.error("updateUser error: " + e.getMessage());
        }
    }

    public List<User> selectAllUser() {
        return new ArrayList<>(userCenter.values());
    }

    public void insertVirtualHost(String username, String virtualHostName) {
        try {
            VirtualHost virtualHost = new VirtualHost(username, virtualHostName);
            virtualHost.startScanThread();
            VirtualHost.VirtualHostGloballyUniqueIdentity virtualHostGloballyUniqueIdentity = new VirtualHost.VirtualHostGloballyUniqueIdentity(username, virtualHostName);
            virtualHostCenter.put(virtualHostGloballyUniqueIdentity, virtualHost);
        } catch (Exception e) {
            log.error("insertVirtualHost error: " + e.getMessage());
        }
    }

    public void deleteVirtualHost(String username, String virtualHostName) {
        try {
            VirtualHost.VirtualHostGloballyUniqueIdentity virtualHostGloballyUniqueIdentity = new VirtualHost.VirtualHostGloballyUniqueIdentity(username, virtualHostName);
            VirtualHost virtualHost = virtualHostCenter.get(virtualHostGloballyUniqueIdentity);
            if (virtualHost == null) {
                // 待删除虚拟主机不存在
                return;
            }
            virtualHostCenter.remove(virtualHostGloballyUniqueIdentity);
        } catch (Exception e) {
            log.error("deleteVirtualHost error: " + e.getMessage());
        }
    }

    public VirtualHost selectVirtualHost(String username, String virtualHostName) {
        try {
            VirtualHost.VirtualHostGloballyUniqueIdentity virtualHostGloballyUniqueIdentity = new VirtualHost.VirtualHostGloballyUniqueIdentity(username, virtualHostName);
            VirtualHost virtualHost = virtualHostCenter.get(virtualHostGloballyUniqueIdentity);
            if (virtualHost == null) {
                // 待查询虚拟主机不存在
                return null;
            }
            return virtualHost;
        } catch (Exception e) {
            // 没有获取到 id
            return null;
        }
    }

    public List<VirtualHost> selectAllVirtualHost() {
        return new ArrayList<>(virtualHostCenter.values());
    }

    public void insertExchange(String username, String virtualHostName, String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete) {
        try {
            Exchange.ExchangeGloballyUniqueIdentity exchangeGloballyUniqueIdentity = new Exchange.ExchangeGloballyUniqueIdentity(username, virtualHostName, exchangeName);
            Exchange exchange = new Exchange(username, virtualHostName, exchangeName, exchangeType, durable, autoDelete, null);
            exchangeCenter.put(exchangeGloballyUniqueIdentity, exchange);

        } catch (Exception e) {
            log.error("insertExchange error: " + e.getMessage());
        }
    }

    public void deleteExchange(String username, String virtualHostName, String exchangeName) {
        try {
            Exchange.ExchangeGloballyUniqueIdentity exchangeGloballyUniqueIdentity = new Exchange.ExchangeGloballyUniqueIdentity(username, virtualHostName, exchangeName);
            Exchange exchange = exchangeCenter.get(exchangeGloballyUniqueIdentity);
            if (exchange == null) {
                // 待删除交换机不存在
                return;
            }
            exchangeCenter.remove(exchangeGloballyUniqueIdentity);
        } catch (Exception e) {
            log.error("deleteExchange error: " + e.getMessage());
        }
    }

    public Exchange selectExchange(String username, String virtualHostName, String exchangeName) {
        try {
            Exchange.ExchangeGloballyUniqueIdentity exchangeGloballyUniqueIdentity = new Exchange.ExchangeGloballyUniqueIdentity(username, virtualHostName, exchangeName);
            Exchange exchange = exchangeCenter.get(exchangeGloballyUniqueIdentity);
            if (exchange == null) {
                // 待查询交换机不存在
                return null;
            }
            return exchange;
        } catch (Exception e) {
            return null;
        }
    }

    public List<Exchange> selectAllExchange() {
        return new ArrayList<>(exchangeCenter.values());
    }

    public void insertQueue(String username, String virtualHostName, String queueName, boolean durable, boolean exclusive, boolean autoDelete) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            Queue queue = new Queue(username, virtualHostName, queueName, durable, exclusive, autoDelete, null);
            queueCenter.put(queueGloballyUniqueIdentity, queue);
        } catch (Exception e) {
            log.error("insertQueue error: " + e.getMessage());
        }
    }

    public void deleteQueue(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            Queue queue = queueCenter.get(queueGloballyUniqueIdentity);
            if (queue == null) {
                // 待删除队列不存在
                return;
            }
            queueCenter.remove(queueGloballyUniqueIdentity);
        } catch (Exception e) {
            log.error("deleteQueue error: " + e.getMessage());
        }
    }

    public Queue selectQueue(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            Queue queue = queueCenter.get(queueGloballyUniqueIdentity);
            if (queue == null) {
                // 待查询队列不存在
                return null;
            }
            return queue;
        } catch (Exception e) {
            return null;
        }
    }

    public List<Queue> selectAllQueue() {
        return new ArrayList<>(queueCenter.values());
    }

    public void insertBinding(String username, String virtualHostName, String exchangeName, String queueName, String bindingKey) {
        try {
            Binding binding = new Binding(username, virtualHostName, exchangeName, queueName, bindingKey);
            Binding.BindingeGloballyUniqueIdentity bindingeGloballyUniqueIdentity = new Binding.BindingeGloballyUniqueIdentity(username, virtualHostName, exchangeName, queueName);
            bindingCenter.put(bindingeGloballyUniqueIdentity, binding);
        } catch (Exception e) {
            log.error("insertBinding error: " + e.getMessage());
        }
    }

    public void deleteBinding(String username, String virtualHostName, String exchangeName, String queueName) {
        try {
            Binding.BindingeGloballyUniqueIdentity bindingeGloballyUniqueIdentity = new Binding.BindingeGloballyUniqueIdentity(username, virtualHostName, exchangeName, queueName);
            Binding binding = bindingCenter.get(bindingeGloballyUniqueIdentity);
            if (binding == null) {
                // 待删除绑定不存在
                return;
            }
            bindingCenter.remove(bindingeGloballyUniqueIdentity);
        } catch (Exception e) {
            log.error("deleteBinding error: " + e.getMessage());
        }
    }


    public Binding selectBinding(String username, String virtualHostName, String exchangeName, String queueName) {
        try {
            Binding.BindingeGloballyUniqueIdentity bindingeGloballyUniqueIdentity = new Binding.BindingeGloballyUniqueIdentity(username, virtualHostName, exchangeName, queueName);
            Binding binding = bindingCenter.get(bindingeGloballyUniqueIdentity);
            if (binding == null) {
                // 待查询绑定不存在
                return null;
            }
            return binding;
        } catch (Exception e) {
            return null;
        }
    }

    public List<Binding> selectAllBinding() {
        return new ArrayList<>(bindingCenter.values());
    }

    public void insertMessageToQueue(String username, String virtualHostName, String queueName, Message message) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.computeIfAbsent(queueGloballyUniqueIdentity, k -> new LinkedList<>());
            messages.add(message);
        } catch (Exception e) {
            log.error("insertMessageToQueue error: " + e.getMessage());
        }
    }

    public boolean deleteMessageFromQueue(String username, String virtualHostName, String queueName, Message message) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.get(queueGloballyUniqueIdentity);
            if (messages == null) {
                // 待删除消息不存在
                return false;
            }
            messages.remove(message);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    public void insertMessageToCenter(Message message) {
        messageCenter.put(message.getMessageId(), message);
    }


    public void deleteMessageFromCenter(Message message) {
        messageCenter.remove(message.getMessageId());
    }

    public Message selectMessageFromCenter(String messageId) {
        return messageCenter.get(messageId);
    }

    public List<Message> selectAllMessageFromCenter() {
        return new ArrayList<>(messageCenter.values());
    }

    public List<Message> selectAllMessageFromQueue(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.get(queueGloballyUniqueIdentity);
            if (messages == null) {
                return new LinkedList<>();
            }
            return messages;
        } catch (Exception e) {
            return new LinkedList<>();
        }
    }

    public Message pollMessage(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.get(queueGloballyUniqueIdentity);
            if (messages == null) {
                return null;
            }
            if (messages.isEmpty()) {
                return null;
            }
            return messages.poll();
        } catch (Exception e) {
            return null;
        }
    }


    public void insertMessageToWaitAckMessageQueue(String username, String virtualHostName, String queueName, Message message) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            ConcurrentHashMap<String, Message> waitAckMessageQueue = waitAckMessageCenter.computeIfAbsent(queueGloballyUniqueIdentity, k -> new ConcurrentHashMap<>());
            waitAckMessageQueue.put(message.getMessageId(), message);
        } catch (Exception e) {
            log.error("insertMessageToWaitAckMessageQueue error: " + e.getMessage());
        }
    }

    public boolean deleteMessageFromWaitAckMessageQueue(String username, String virtualHostName, String queueName, Message message) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            ConcurrentHashMap<String, Message> waitAckMessageQueue = waitAckMessageCenter.get(queueGloballyUniqueIdentity);
            if (waitAckMessageQueue == null) {
                // 待删除消息不存在
                return false;
            }
            waitAckMessageQueue.remove(message.getMessageId());
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    public Message selectMessageFromWaitAckMessageQueue(String username, String virtualHostName, String queueName, String messageId) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);

            ConcurrentHashMap<String, Message> waitAckMessageQueue = waitAckMessageCenter.get(queueGloballyUniqueIdentity);
            if (waitAckMessageQueue == null) {
                return null;
            }
            return waitAckMessageQueue.get(messageId);
        } catch (Exception e) {
            return null;
        }
    }

    public List<Binding> selectBindingsFromExchange(String username, String virtualHostName, String exchangeName) {
        try {
            return bindingCenter.values().stream().filter(binding -> (binding.getUsername().equals(username)) && (binding.getVirtualHostName().equals(virtualHostName)) && (binding.getExchangeName().equals(exchangeName))).collect(Collectors.toList());

        } catch (Exception e) {
            return new ArrayList<>();
        }
    }

    public List<VirtualHost> selectVirtualHostByUser(String username) {
        try {
            return virtualHostCenter.values().stream().filter(virtualHost -> virtualHost.getUsername().equals(username)).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }
    }

    public List<Exchange> selectExchangeByVirtualHost(String username, String virtualHostName) {
        try {
            return exchangeCenter.values().stream().filter(exchange -> (exchange.getUsername().equals(username)) && (exchange.getVirtualHostName().equals(virtualHostName))).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }
    }

    public List<Queue> selectQueueByVirtualHost(String username, String virtualHostName) {
        try {
            return queueCenter.values().stream().filter(queue -> (queue.getUsername().equals(username)) && (queue.getVirtualHostName().equals(virtualHostName))).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }
    }

    public List<Binding> selectBindingByVirtualHost(String username, String virtualHostName) {
        try {
            return bindingCenter.values().stream().filter(binding -> (binding.getUsername().equals(username)) && (binding.getVirtualHostName().equals(virtualHostName))).collect(Collectors.toList());
        } catch (Exception e) {
            return null;
        }

    }

    public List<Message> selectMessageByVirtualHost(String username, String virtualHostName) {
        try {
            List<Message> messages = new LinkedList<>();
            // 获取虚拟机的所有队列
            List<Queue> queueList = selectQueueByVirtualHost(username, virtualHostName);
            for (Queue queue : queueList) {
                messages.addAll(Objects.requireNonNull(selectMessageByQueue(username, virtualHostName, queue.getQueueName())));
            }

            return messages;
        } catch (Exception e) {
            return null;
        }
    }

    private List<Message> selectMessageByQueue(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.get(queueGloballyUniqueIdentity);
            if (messages == null) {
                return new LinkedList<>();
            }
            return messages;
        } catch (Exception e) {
            return null;
        }
    }

    public int getQueueMessageCount(String username, String virtualHostName, String queueName) {
        try {
            Queue.QueueGloballyUniqueIdentity queueGloballyUniqueIdentity = new Queue.QueueGloballyUniqueIdentity(username, virtualHostName, queueName);
            LinkedList<Message> messages = queueMessageCenter.get(queueGloballyUniqueIdentity);
            if (messages == null) {
                return 0;
            }
            return messages.size();
        } catch (Exception e) {
            return 0;
        }
    }
}
